#include "brpc.h"
#include "kafka_consummer.hpp"
#include <algorithm>
#include <brpc/naming_service.h>
#include <gflags/gflags.h>
#include <memory>
#include "behavior_analytics_server.h"

// reids
DEFINE_string(redis_host,"localhost", "redis连接本地");
DEFINE_int32(redis_port, 6379,"数据库的启动端口");
DEFINE_int32(redis_db, 0, "本地数据库默认启动id=0");
DEFINE_bool(redis_keep_alive, true, "防止空闲连接被断开");
// blog behavior
DEFINE_string(blog_action_broker_host, "localhost:9092", "Kafka的broker地址列表");
DEFINE_string(blog_action_topic_name, "blog-action", "topic名称");
DEFINE_string(blog_action_group_id,"blog-action-group","消费者组名");

int main(){
    lyt::BehaviorAnalytics::ptr behavior_analytics = std::make_shared<lyt::BehaviorAnalytics>\
    (FLAGS_redis_host, FLAGS_redis_port, FLAGS_redis_db,FLAGS_redis_keep_alive);
    lyt::KafkaConsumer::ptr blogview_consumer = std::make_unique<lyt::KafkaConsumer>(FLAGS_blog_action_broker_host
                                                                                ,FLAGS_blog_action_group_id
                                                                                ,FLAGS_blog_action_topic_name);
    // 注册回调
    blogview_consumer->setMessageCallback(
        [&](std::unique_ptr<RdKafka::Message> msg){
        behavior_analytics->BehaviorCallBack(std::move(msg));
    }).start();
    
    return 0;
}